package cn.yi.boot.mq.rocketmq.listener;

import cn.yi.boot.mq.coder.MessageDecoder;
import cn.yi.boot.mq.rocketmq.message.handler.MessageHandler;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.Map;

public class DefaultMessageListener implements MessageListenerConcurrently {

    @Setter
    @Getter
    private Map<String, Map<String, MessageHandler>> topic2TagHandler;


    public void init() {
        if (null == topic2TagHandler || topic2TagHandler.isEmpty()) {
            throw new RuntimeException("未配置相关的Topic方案");
        }
        topic2TagHandler.forEach((topic, tagHandler) -> {
            if (null == tagHandler || tagHandler.isEmpty()) {
                throw new RuntimeException("未配置相关的Tag方案");
            }
        });
    }

    /**
     * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
     *
     * @param list
     * @param consumeConcurrentlyContext
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt messageExt : list) {
            String topic = messageExt.getTopic();
            Map<String, MessageHandler> tagHandler = topic2TagHandler.get(topic);
            if (null == tagHandler || tagHandler.isEmpty()) {
                tagHandler = topic2TagHandler.get("*");
            }
            if (null == tagHandler || tagHandler.isEmpty()) {
                throw new RuntimeException("未配置相关的Tag方案");
            }

            MessageHandler messageHandler = null;
            String tag = messageExt.getTags();
            if (StringUtils.isEmpty(tag)) {
                messageHandler = tagHandler.get("*");
            } else {
                messageHandler = tagHandler.get(tag);
            }
            MessageDecoder messageDecoder = messageHandler.getMessageDecoder();
            return messageHandler.process(messageDecoder.decode(messageExt.getBody()), messageExt, consumeConcurrentlyContext);

        }

        return null;
    }
}
